package com.wp.spark

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}


//不带offset控制的版本

object SparkKafka {
  def main(args: Array[String]): Unit = {
    //创建环境
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("testApp")

    val ssc = new StreamingContext(conf, Seconds(1))
    val sc: SparkContext = ssc.sparkContext
    sc.setLogLevel("ERROR")


    //业务处理
    val topics = Set("test")
    val kafkaParamsmap = Map(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "cdh141:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "kafka_Source",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer"
    )

    val kafkaDSStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParamsmap))
    val valueDSStream: DStream[String] = kafkaDSStream.map(record => record.value())
    valueDSStream.foreachRDD(rdd=>{
      rdd.foreach(println)
    })
    valueDSStream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).foreachRDD(rdd=>{
      rdd.collect().foreach(println)
    })
    //这个会打印时间戳
    //启动采集器
    ssc.start()
    ssc.awaitTermination()
  }
}
